package app.function;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import app.bean.SearchRankResult;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;

/**
 * 搜索排名处理函数 - 实现实时排名计算
 */
public class SearchRankProcessFunction extends KeyedProcessFunction<String, SearchRankResult, SearchRankResult> {
    
    private final String rankType;
    private final int topN;
    private ListState<SearchRankResult> searchListState;
    private static final String DATA_SOURCE = "kafka_dwd_market_search";
    private static final Integer VERSION = 1;
    
    public SearchRankProcessFunction(String rankType, int topN) {
        this.rankType = rankType;
        this.topN = topN;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        searchListState = getRuntimeContext().getListState(
            new ListStateDescriptor<>("searchListState", 
                TypeInformation.of(new TypeHint<SearchRankResult>() {}))
        );
    }
    
    @Override
    public void processElement(SearchRankResult value, Context ctx, Collector<SearchRankResult> out) throws Exception {
        // 添加到状态列表
        searchListState.add(value);
        
        // 注册定时器，延迟1秒处理排名
        ctx.timerService().registerEventTimeTimer(ctx.timerService().currentWatermark() + 1000);
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<SearchRankResult> out) throws Exception {
        // 获取所有搜索数据
        List<SearchRankResult> searchList = new ArrayList<>();
        for (SearchRankResult search : searchListState.get()) {
            searchList.add(search);
        }
        
        // 根据排名类型进行排序
        switch (rankType) {
            case "search_hot":
                searchList.sort(Comparator.comparing(SearchRankResult::getSearchCount, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "search_transaction":
                // SearchRankResult没有getTotalSalesAmount字段，使用orderCount作为交易排名依据
                searchList.sort(Comparator.comparing(SearchRankResult::getOrderCount, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "search_click":
                searchList.sort(Comparator.comparing(SearchRankResult::getClickCount, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "search_conversion":
                searchList.sort(Comparator.comparing(SearchRankResult::getConversionRate, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "blue_ocean":
                searchList.sort(Comparator.comparing(SearchRankResult::getDemandSupplyRatio, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            default:
                searchList.sort(Comparator.comparing(SearchRankResult::getSearchCount, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
        }
        
        // 输出前TopN搜索词，设置排名和标准字段
        for (int i = 0; i < Math.min(topN, searchList.size()); i++) {
            SearchRankResult search = searchList.get(i);
            
            // 设置排名信息
            search.setRank(i + 1);
            search.setRankType(rankType);
            
            // 设置实时数仓标准字段
            search.setCurrentTimestamp();
            search.setDataLineage(DATA_SOURCE, VERSION);
            search.setPeriod("1d"); // 1天统计周期
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            long windowStart = timestamp - 86400000; // 1天前
            search.setWindowTime(sdf.format(new Date(windowStart)), sdf.format(new Date(timestamp)));
            
            out.collect(search);
        }
        
        // 清空状态
        searchListState.clear();
    }
}